package cn.jolyoulu.mqtt.service;


import cn.jolyoulu.mqtt.entity.dto.MqttMsgDTO;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.*;
import org.springframework.boot.autoconfigure.condition.ConditionalOnBean;
import org.springframework.stereotype.Component;
import cn.jolyoulu.common.core.queue.delayed.DelayedTaskEntity;
import cn.jolyoulu.common.core.queue.delayed.observer.AbstractObserver;
import cn.jolyoulu.common.core.queue.delayed.observer.DelayedTaskSubject;
import cn.jolyoulu.mqtt.config.mqtt.EnableMqttClient;
import cn.jolyoulu.mqtt.config.mqtt.MqttClientPool;

import java.time.Duration;
import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Objects;

@Slf4j
@Component
@ConditionalOnBean(annotation = EnableMqttClient.class)
public class MqttPushObserver extends AbstractObserver<MqttMsgDTO> {

    /**
     * 发布消息
     */
    public String pubTopicPrefix = "testtopic/";

    /**
     * QoS 0表示最多一次传递（At Most Once），消息推送出去后并不检查是否到达。
     * QoS 1表示至少一次传递（At Least Once），消息推送后会检查服务端/设备端返回的Ack，若超时则会重发。能保证对方收到消息，但可能存在消息重复。
     * QoS 2表示恰好一次传递（Exactly Once）。 选择较高的QoS级别可以提高消息传递的可靠性，但同时也会增加通信的延迟和带宽消耗，并增大对服务器资源的需求。
     */
    private Integer qos = 2;

    /**
     * 客户端连接池
     */
    private MqttClientPool pool;

    /**
     * 你需要观察的那个对象Class
     * 如：super(GenCodeFormBO.class);
     */
    protected MqttPushObserver(DelayedTaskSubject delayedTaskSubject, MqttClientPool pool) {
        super(MqttMsgDTO.class);
        this.pool = pool;
        delayedTaskSubject.registerObserver(this);
    }

    @Override
    protected void task(DelayedTaskEntity<MqttMsgDTO> delayedTask) {
        //获取一个连接
        MqttClient client = null;
        try {
            client = pool.getClient();
            MqttMsgDTO task = delayedTask.getTask();
            MqttMessage message = new MqttMessage(task.toJsonString().getBytes());
            message.setQos(qos);
            //判断是否是延迟消息
            LocalDateTime sendDateTime = LocalDateTime.parse(task.getSendTime(), DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
            if (LocalDateTime.now().equals(sendDateTime) || LocalDateTime.now().isAfter(sendDateTime)){
                client.publish(pubTopicPrefix + task.getToDeviceSn(), message);
            }else {
                //计算延迟秒数
                Duration between = Duration.between(LocalDateTime.now(), sendDateTime);
                client.publish("$delayed/"+between.getSeconds()+"/"+pubTopicPrefix + task.getToDeviceSn(), message);
            }
        } catch (Exception e) {
            log.error("消息发布失败", e);
        } finally {
            if (Objects.nonNull(client)) {
                pool.returnClient(client);
            }
        }
    }
}
